package com.maben.mqtt.config;

import com.maben.mqtt.constant.YmlConstant;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import javax.annotation.Resource;
import java.util.Objects;

/**
 * 生产端配置
 */
@Configuration
public class MqttProviderConfig {

    @Resource
    private YmlConstant ymlConstant;
    /**
     * 发布的bean名称
     */
    public static final String CHANNEL_NAME_OUT = "mqttOutboundChannel";


    /**
     * MQTT连接器选项
     */
    @Bean
    public MqttConnectOptions getMqttConnectOptions() {
        //连接设置
        MqttConnectOptions options = new MqttConnectOptions();
        //设置url
        options.setServerURIs(StringUtils.split(ymlConstant.getProviderHostUrl(), ","));
        //设置连接用户名
        options.setUserName(ymlConstant.getProviderUsername());
        //设置连接密码
        options.setPassword(ymlConstant.getProviderPassword().toCharArray());
        //是否清空session，设置为false表示服务器会保留客户端的连接记录（订阅主题，qos），客户端重连之后能获取到服务器在客户端断开连接期间推送的消息
        //设置为true表示每次连接到服务端都是以新的身份
        options.setCleanSession(ymlConstant.getProviderCleanSession());
        //设置超时时间，单位为秒
        options.setConnectionTimeout(ymlConstant.getProviderConnectTimeout());
        //设置心跳时间 单位为秒，表示服务器每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线
        options.setKeepAliveInterval(ymlConstant.getProviderKeepAliveInterval());
        //设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息
        options.setWill("willTopic", (ymlConstant.getProviderClientId() + "与服务器断开连接").getBytes(), 0, false);
        return options;
    }

    /**
     * MQTT客户端
     */
    @Bean
    public MqttPahoClientFactory mqttProviderFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(getMqttConnectOptions());
        return factory;
    }

    /**
     * MQTT信息通道（生产者）
     */
    @Bean(name = CHANNEL_NAME_OUT)
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    /**
     * MQTT消息处理器（生产者）
     */
    @Bean
    @ServiceActivator(inputChannel = CHANNEL_NAME_OUT)
    public MessageHandler mqttOutbound() {
        //设置clientId
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(ymlConstant.getProviderClientId(), mqttProviderFactory());
        //设置异步
        messageHandler.setAsync(true);
        if (Objects.nonNull(ymlConstant.getProviderRetained())) {
            messageHandler.setDefaultRetained(ymlConstant.getProviderRetained());
        }
        if (StringUtils.isNotBlank(ymlConstant.getProviderDefaultTopic())) {
            messageHandler.setDefaultTopic(ymlConstant.getProviderDefaultTopic());
        }
        return messageHandler;
    }
}

